package com.ddyx.service.impl;

import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.ddyx.service.IMessageQueueService;

//@Service("messageQueueService")
public class AliyunKafkaMessageQueueService implements IMessageQueueService {

	private static final Logger logger = LoggerFactory.getLogger(AliyunKafkaMessageQueueService.class);
	
	private KafkaProducer<String, String> producer;

	public AliyunKafkaMessageQueueService() {
		if (System.getProperty("java.security.auth.login.config") == null) {
			System.setProperty("java.security.auth.login.config", AliyunKafkaMessageQueueService.class.getClassLoader().getResource("kafka_client_jaas.conf").getPath());
		}

		Properties props = new Properties();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "SASL_SSL://kafka-ons-internet.aliyun.com:8080");
		props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, AliyunKafkaMessageQueueService.class.getClassLoader().getResource("kafka.client.truststore.jks").getPath());
		props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
		props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
		props.put(SaslConfigs.SASL_MECHANISM, "ONS");
		props.put(ProducerConfig.RETRIES_CONFIG, 0);
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");
		props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
		producer = new KafkaProducer<String, String>(props);
	}

	@Override
	public void publish(String topic, String message) {
		try {
			Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<String, String>(topic, null,
					System.currentTimeMillis(), String.valueOf(message.hashCode()), message));
			RecordMetadata recordMetadata = metadataFuture.get();
			logger.info("Produce Sucessed: {}", recordMetadata.toString());
		} catch (Exception err) {
			logger.error("Produce failed", err);
		}
		producer.flush();
	}
}
